package com.gitee.huxing.course.controller;

import com.gitee.huxing.course.vo.CreateTopicDTO;
import com.gitee.huxing.course.vo.SendMsgDTO;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;

/**
 * @Description: kafka功能测试类
 * @ProjectName: gdca-platform
 * @Package: com.gdca.cert.controller
 * @ClassName: KafkaTestController
 * @Author: huxing
 * @DateTime: 2021-08-23 下午3:16
 */
@RestController
public class KafkaTestController {

    @Autowired
    private KafkaProducer kafkaProducer;

    @Autowired
    private KafkaAdmin kafkaAdmin;

    /**
     * @Description:
     * @Author: huxing
     * @param sendMsgDTO
     * @return java.lang.String
     * @Date: 2021/8/23 下午2:59
     **/
    @ResponseBody
    @PostMapping("/api/v1/kafka/send")
    public String send(SendMsgDTO sendMsgDTO){
        kafkaProducer.send(sendMsgDTO.getTopic(),
                sendMsgDTO.getBody());
        return "消息发送成功";
    }

    /**
     * @Description:
     * @Author: huxing
     * @param topicDTO 订阅主题对象
     * @return java.lang.String
     * @Date: 2021/8/24 下午2:22
     **/
    @ResponseBody
    @PostMapping("/api/v1/kafka/add")
    public String createTopic(CreateTopicDTO topicDTO){
        // 第一个参数对应 订阅主题 topic
        // 第二个参数对应 partitions 创建的节点 默认用1
        // 第三个参数对应 创建的节点  用3 代表是集群片区数量
        AdminClient client = AdminClient.create(kafkaAdmin.getConfigurationProperties());
        NewTopic kafkaTopic = new NewTopic(topicDTO.getTopic(), 1, (short) 3);
        CreateTopicsResult result = client.createTopics(Arrays.asList(kafkaTopic));
        return result.toString();
    }
}
